package com.chukun.multi.activeobject;

import org.apache.log4j.Logger;

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author chukun
 * 模式角色 : ActiveObject.Proxy
 */
public class AsyncRequestPersistence implements RequestPersistence {

    private static final long ONE_MINUTE_IN_SECOND = 60;
    private final Logger logger;
    private final AtomicInteger taskTimeConsumedPerInterval = new AtomicInteger(0);
    private final AtomicInteger requestSubmittedPerInterval = new AtomicInteger(0);

    private final RequestPersistence delegate = new DiskbasedRequestPersistence();

    // 模式角色 ActiveObject.Scheduler
    private final ThreadPoolExecutor scheduler;

    // 单例模式
    private static class InstanceHolder {
        final static RequestPersistence INSTANCE = new AsyncRequestPersistence();
    }

    private AsyncRequestPersistence() {
        logger = Logger.getLogger(AsyncRequestPersistence.class);
        scheduler = new ThreadPoolExecutor(1, 3,
                60 * ONE_MINUTE_IN_SECOND, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(200),
                r -> {
                    Thread t = new Thread(r,"AsyncRequestPersistence");
                    return t;
                });
        scheduler.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        // 启动队列监控定时任务
        Timer monitorTimer = new Timer(true);
        monitorTimer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                if (logger.isInfoEnabled()) {
                    logger.info("task count: "+requestSubmittedPerInterval.get()
                            +",queue size : "+scheduler.getQueue().size()
                            +",taskTimeConsumedPerInterval : "
                            +taskTimeConsumedPerInterval.get()+" ms");
                }
                taskTimeConsumedPerInterval.set(0);
                requestSubmittedPerInterval.set(0);
            }
        },0L,ONE_MINUTE_IN_SECOND*1000);
    }

    // 返回 AsyncRequestPersistence的唯一实例
    public static RequestPersistence getInstance () {
        return InstanceHolder.INSTANCE;
    }


    @Override
    public void store(MMSDeliveryRequest request) {

        /**
         * 将对store方法的调用封装为 methodRequest 对象，并存入缓冲区
         */
        Callable<Boolean> methodRequest = new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                long start = System.currentTimeMillis();
                try {
                    delegate.store(request);
                }finally {
                    taskTimeConsumedPerInterval.addAndGet((int)(System.currentTimeMillis()-start));
                }
                return Boolean.TRUE;
            }
        };
        scheduler.submit(methodRequest);
        requestSubmittedPerInterval.incrementAndGet();
    }
}
